package com.kamistoat.rocketmqlearning.msgConsumer;

import org.apache.rocketmq.client.consumer.AllocateMessageQueueStrategy;
import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.consumer.listener.*;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.common.consumer.ConsumeFromWhere;
import org.apache.rocketmq.common.message.MessageExt;
import org.apache.rocketmq.common.message.MessageQueue;
import org.apache.rocketmq.common.protocol.heartbeat.MessageModel;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;

import java.util.Arrays;
import java.util.List;
import java.util.concurrent.atomic.AtomicLong;

@Configuration
public class MsgConsumerConfig {
    /**
     * 显然，这是一个 PUSH 模式的消费者。
     * 但实际上RocketMQ的消费者底层使用长轮询POLL模式拉取消息，只是由于POLL零间隔且存在长轮询特性，所以起名叫PUSH
     */
    @Bean
    @Qualifier("generalConsumer")
    public DefaultMQPushConsumer generalConsumer() {
        /**
         * 创建库类Consumer。对其配置进行定制
         */
        DefaultMQPushConsumer generalConsumer = new DefaultMQPushConsumer();
        /**
         * 设置所属消费者组名称。相同消费者组必须以完全相同的消费逻辑存在
         * 默认 DEFAULT_CONSUMER
         */
        generalConsumer.setConsumerGroup("general_consume_group");
        /**
         * 指定NameServer地址。如果NameServer是集群则需使用;指定链接所有的NameServer
         * 生产者会随机链接一个NameServer进行后序流程
         */
        generalConsumer.setNamesrvAddr("172.19.0.20:9876;172.19.0.21:9877");

        /**
         * 设置消费模型，影响队列和消费者的消费关系。仅支持集群消费和广播消费
         * 默认为集群消费CLUSTERING，同一消费者组所有消费者均分消息。广播消费BROADCASTING则是所有消费者都会消费消息
         */
        generalConsumer.setMessageModel(MessageModel.CLUSTERING);
        /**
         * 消费者启动后，从队列的什么位置开始消费。默认是从上次消费的位置开始消费。
         * 注意，由于RocketMQ自动过期机制的存在，消费者可能发现上次消费的位置已经被过期，则此时会自动更改为从当前队列的MIN_OFFSET即第一条未过期消息开始消费。
         */
        generalConsumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_LAST_OFFSET);
        /**
         * Rebalance算法实现策略
         * generalConsumer.setAllocateMessageQueueStrategy()
         */
        /**
         * 消费线程池最小线程数.消费线程池总是存活的线程数量
         * 默认值为10
         */
        generalConsumer.setConsumeThreadMin(10);
        /**
         * 消费线程池最大线程数.当线程空闲时会销毁超过最小线程数的线程
         * 默认值为20
         */
        generalConsumer.setConsumeThreadMax(20);
        /**
         * 消息拉取时间间隔。在Push模式下没有什么作用
         * setPullInterval();
         */
        /**
         * 一个消费线程消费多少条消息。该值决定了消费者开辟多少个线程去消费。不建议设置超过1
         * setConsumeMessageBatchMaxSize();
         * 默认值为1
         */
        /**
         * 批量拉取消息，一次最多拉取多少条消息
         * setPullBatchSize();
         * 默认值为32
         */
        /**
         * 批量拉取的消息会缓存在本地的缓冲队列中，对缓冲队列进行设置: 消息队列的个数 / 消息Topic的个数 / 一个队列中消息的最大数量 / 一个Topic中消息的最大数量
         * setPullThresholdForQueue();
         * setPullThresholdForTopic();
         * setPullThresholdForSizeQueue();
         * setPullThresholdForSizeTopic();
         */

        try {
            /**
             * 建立消费者和Topic的订阅关系。具体和Topic下的哪个队列建立消费关系，和上面的ReBalance策略有关，这将由消费者自己完成队列选择
             * @param someTopic 指定Topic的名称
             * @param * 指定Tags。可以使用&&和||运算符，例如 "Tag1 || Tag2 || Tag3" 的方式，为null或*代表所有Tag
             */
            generalConsumer.subscribe("someTopic", "*");

            /**
             * 注册消息监听器，具体的业务逻辑在监听器中完成处理。
             * MessageListenerConcurrently()接口代表并发消费接口，消费者将使用多个消费线程异步消费消息.
             * 使用多少个消费线程由上面的 setConsumeThreadMin() 决定.
             * 每个消费线程消费多少条消息由上面的 setConsumeMessageBatchMaxSize() 决定.
             */
            generalConsumer.setMessageListener(new MessageListenerConcurrently() {
                /**
                 * 原理讲过，消费者基于长轮询到Broker中POLL消息。
                 * 一旦该消费者订阅的Topic中有消息了，就会触发该<b>异步<b/>监听器方法。消费者从Broker中POLL到消息，消费者从消费线程池中选择一个消费线程(优先使用最小线程池内的线程)进行消费。
                 *
                 * @param msgs 当前消费线程要消费的消息列表。当Topic中存在消息积压时，一个消费线程就可能会处理多条消息。最大值为ConsumeMessageBatchMaxSize，可以在上方进行设置。
                 *             在系统健康的情况下，msgs应在大部分时间内为size=1的列表。因为长轮询下，Broker一有消息就会被消费者POLL到，所以每次消费者也只获得最新的一条消息。
                 *             P.S. 不建议将ConsumeMessageBatchMaxSize的值>1，因为一个消费线程内如果消费多条消息，其中一条出了问题是无法通知Broker的，只能通知这一批消息整体失败。
                 * @param context 消费者上下文。用于记录当前消费线程的相关信息，并在回传Broker时填充必要消息
                 * @return 当前消息的消费状态。将被返回给Broker
                 */
                @Override
                public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs, ConsumeConcurrentlyContext context) {
                    /**
                     * 观察输出。该异步方法会将拉取到的消息分配给一个消费线程来执行
                     */
                    System.out.println("拉取消息:" + msgs.size() + ",承接线程:" + Thread.currentThread());

                    /**
                     * msgs就是当前消费线程需要处理的消息。至于之后怎么消费，同步串行处理还是异步处理，那是你的事，不是RocketMQ关心的。
                     * 业务逻辑消费消息
                     */
                    for (MessageExt msg : msgs) {
                        System.out.println(msg);
                        String s = Arrays.toString(msg.getBody());
                    }

                    /**
                     * 返回该消费线程消费成功给Broker，Broker将会<b>逻辑删除<b/>成功消费的消息。
                     * 不需要进行额外配置，自动使用ConsumeConcurrentlyContext填充必要消息。
                     * P.S. 只能一次性通知当前消费线程消费的ConsumeMessageBatchMaxSize条msgs成功与否，因此不建议一个消费线程的ConsumeMessageBatchMaxSize大于1，否则无法定位失败消息。
                     */
                    return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
                }
            });

            // 启动消费者
            generalConsumer.start();
        } catch (MQClientException e) {
            e.printStackTrace();
        }

        return generalConsumer;
    }

    /**
     * 顺序消息消费者。
     */
    @Bean
    @Qualifier("orderConsumer")
    public DefaultMQPushConsumer orderConsumer() {
        DefaultMQPushConsumer orderConsumer = new DefaultMQPushConsumer();
        orderConsumer.setConsumerGroup("order_consume_group");
        orderConsumer.setNamesrvAddr("172.19.0.20:9876;172.19.0.21:9877");
        orderConsumer.setMessageModel(MessageModel.CLUSTERING);
        orderConsumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_LAST_OFFSET);
        orderConsumer.setConsumeThreadMin(10);
        orderConsumer.setConsumeThreadMax(20);

        try {
            orderConsumer.subscribe("orderTopic", "*");
            /**
             * 如果要进行顺序消费需要使用 MessageListenerOrderly()顺序消费接口。
             * 一旦长轮询POLL到消息就回调下方的方法进行消费
             */
            orderConsumer.setMessageListener(new MessageListenerOrderly() {
                /**
                 * MessageListenerOrderly() 需要覆写的 consumeMessage() 方法
                 * @param msgs POLL到的消息。在顺序消费下，Broker保证每次投递一条消息
                 * @param context 又Broker自动填充
                 * @return
                 */
                @Override
                public ConsumeOrderlyStatus consumeMessage(List<MessageExt> msgs, ConsumeOrderlyContext context) {
                    System.out.printf("%s Receive New Messages: %s %n", Thread.currentThread().getName(), new String(msgs.get(0).getBody()));
                    return ConsumeOrderlyStatus.SUCCESS;
                }
            });
            // 启动消费者，在该Bean中启动一次即可。
            orderConsumer.start();
        } catch (MQClientException e) {
            e.printStackTrace();
        }

        return orderConsumer;
    }

    /**
     * 延时消息消费者
     * @return
     */
    @Bean
    @Qualifier("delayConsumer")
    public DefaultMQPushConsumer delayConsumer() {
        DefaultMQPushConsumer delayConsumer = new DefaultMQPushConsumer();
        delayConsumer.setConsumerGroup("delay_consume_group");
        delayConsumer.setNamesrvAddr("172.19.0.20:9876;172.19.0.21:9877");
        delayConsumer.setMessageModel(MessageModel.CLUSTERING);
        delayConsumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_LAST_OFFSET);
        delayConsumer.setConsumeThreadMin(10);
        delayConsumer.setConsumeThreadMax(20);

        try {
            delayConsumer.subscribe("delayTopic", "*");
            /**
             * 如果要进行顺序消费需要使用 MessageListenerOrderly 接口，而不能使用普通消息的监听器
             */
            delayConsumer.setMessageListener(new MessageListenerConcurrently() {
                @Override
                public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs, ConsumeConcurrentlyContext context) {
                    System.out.printf("%s Receive New Messages: %s %n", Thread.currentThread().getName(), new String(msgs.get(0).getBody()));
                    return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
                }
            });
            // 启动消费者
            delayConsumer.start();
        } catch (MQClientException e) {
            e.printStackTrace();
        }
        return delayConsumer;
    }
}
